package com.fudan.run.action;

import java.util.Map;

import org.apache.spark.api.java.JavaRDD;

import com.fudan.cfg.RunConfiguration;
import com.fudan.cfg.action.Foreach;
import com.fudan.ctx.EnvCtx;
import com.fudan.run.VariableRddMap;
import com.fudan.run.ctx.annotation.ActionAdapter;

@ActionAdapter("foreach")
public class ForeachAction extends ActionHandler<Foreach>{

	@Override
	public void action(Foreach actionNode, VariableRddMap variableRddMap,RunConfiguration runConfiguration) {
		JavaRDD<Map<String, Object>> target = variableRddMap.getRdd(actionNode.getTarget());
		target.foreach(l->{
			System.out.println(l);
			EnvCtx.envOut.println(l);
		});
	}
}
